package com.hgw.easyrocketmq.transaction;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

/**
 * Description: 事务消息生产者
 *
 * @author LinHuiBa-YanAn
 * @date 2023/2/15 15:14
 */
public class TransactionProducer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        // 创建消息生产者
        TransactionMQProducer mqProducer = new TransactionMQProducer("group2");
        mqProducer.setNamesrvAddr("47.96.232.192:9876");
        // 编写监听器逻辑
        mqProducer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                System.out.println("执行本地事务");
                if (StringUtils.equals("TagA", msg.getTags())) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if (StringUtils.equals("TagB", msg.getTags())) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else {
                    return LocalTransactionState.UNKNOW;
                }
            }
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("MQ检查消息Tag【" + msg.getTags() + "】的本地事务执行结果");
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        // 启动消息生产者
        mqProducer.start();
        String[] tags = {"TagA", "TagB", "TagC"};
        for (int i = 0; i < 3; i++) {
            Message msg = new Message("TransactionTopic", tags[i % tags.length], "KEY" + i, ("Hello，RocketMQ" + i).getBytes(StandardCharsets.UTF_8));
            TransactionSendResult sendResult = mqProducer.sendMessageInTransaction(msg, null);
            System.out.println(sendResult);
            TimeUnit.SECONDS.sleep(1);
        }
        mqProducer.shutdown();
    }
}
